package com.xcf.client;

import com.xcf.center.ServerMeta;
import com.xcf.center.client.IListener;
import com.xcf.center.client.ServerManager;
import com.xcf.center.client.ServerWrapper;
import com.xcf.codec.RequestJsonDecoder;
import com.xcf.codec.RequestJsonEncoder;
import com.xcf.model.RpcRequest;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class RpcConsumer implements IListener {

    //创建线程池对象
    private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private static Timer timer = new Timer("Server-Balance-Check", false);

    private static volatile UserClientHandler userClientHandler;

    private Map<String, Map<String, UserClientHandler>> clientHandlerMap = new ConcurrentHashMap<>();

    private volatile List<ServerWrapper<ServerMeta>> serverList = new ArrayList<>();

    private final ServerManager serverManager;

    private volatile Integer state;

    private static final Object serverLock = new Object();

    public RpcConsumer(ServerManager serverManager) {
        this.serverManager = serverManager;
        serverManager.setListener(this);
        state = 0;
    }

    @Override
    public synchronized void onChange(String nameSpace, List<ServerWrapper<ServerMeta>> serverList) {
        Map<String, UserClientHandler> newMap = new HashMap<>();
        Map<String, UserClientHandler> oldMap = clientHandlerMap.getOrDefault(nameSpace, new HashMap<>());
        for (ServerWrapper<ServerMeta> serverWrapper : serverList) {
            String uniqueName = serverWrapper.getServerMeta().getUniqueName();
            UserClientHandler oldHandler = oldMap.get(uniqueName);
            if (null != oldHandler) {
                newMap.put(uniqueName, oldHandler);
            } else {
                // 新建
                UserClientHandler clientHandler = null;
                try {
                    clientHandler = initClient(serverWrapper);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                if (null != clientHandler) {
                    newMap.put(uniqueName, clientHandler);
                }
            }
        }
        synchronized (serverLock) {
            this.serverList = serverList;
        }
        clientHandlerMap.put(nameSpace, newMap);
        if (state == 0) {
            timer.schedule(new ServerTimeCheckTask(), 1, 5000);
            state = 1;
        }
        System.out.println("Init client finish.");
    }

    //1.创建一个代理对象 providerName：UserService#sayHello are you ok?
    public <T> T createProxy(String nameSpace, final Class<T> serviceClass, final String beanName) {
        //借助JDK动态代理生成代理对象
        return (T) Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[]{serviceClass}, new InvocationHandler() {
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                //（1）调用初始化netty客户端的方法
                String methodName = method.getName();
                if ("hashCode".equals(methodName) || "equals".equals(methodName)
                        || "clone".equals(methodName) || "toString".equals(methodName)
                        || "notify".equals(methodName) || "notifyAll".equals(methodName)
                        || "wait".equals(methodName) || "finalize".equals(methodName)) {
                    return method.invoke(proxy, args);
                }
                if (state == 0) {
                    throw new IllegalStateException("还未初始化过客户端");
                }
                UserClientHandler handler = getHandler(nameSpace);
                RpcRequest request = new RpcRequest();
                request.setRequestId(beanName);
                request.setTargetClassName(serviceClass.getName());
                request.setTargetClass(serviceClass);
                request.setMethodName(methodName);
                request.setParameters(args);
                if (null != args) {
                    Class<?>[] agrTypes = new Class<?>[args.length];
                    for (int i = 0; i < args.length; i++) {
                        agrTypes[i] = args[i].getClass();
                    }
                    request.setParameterTypes(agrTypes);
                }
                // 设置参数
                handler.setRequest(request);
                Class<?> returnType = method.getReturnType();
                handler.setResultType(returnType);
                // 去服务端请求数据
                return executor.submit(handler).get();
            }
        });
    }

    private synchronized UserClientHandler getHandler(String nameSpace) {
        ServerWrapper<ServerMeta> wrapper = serverManager.getServer(nameSpace);
        if (wrapper == null) {
            throw new NullPointerException("No server can be provider service.");
        }
        Map<String, UserClientHandler> handlerMap = clientHandlerMap.get(nameSpace);
        UserClientHandler handler = handlerMap.get(wrapper.getServerMeta().getUniqueName());
        handler.setServerWrapper(wrapper);
        return handler;
    }


    //2.初始化netty客户端
    public static void initClient() throws InterruptedException {
        userClientHandler = new UserClientHandler();
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(65535))
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new RequestJsonEncoder(userClientHandler));
                        pipeline.addLast(new RequestJsonDecoder(userClientHandler));
                        pipeline.addLast(userClientHandler);
                    }
                });
        bootstrap.connect("127.0.0.1", 8990).sync();
    }

    //2.初始化netty客户端
    public static UserClientHandler initClient(ServerWrapper<ServerMeta> serverWrapper) throws InterruptedException {
        UserClientHandler userClientHandler = new UserClientHandler();
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(65535))
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new RequestJsonEncoder(userClientHandler));
                        pipeline.addLast(new RequestJsonDecoder(userClientHandler));
                        pipeline.addLast(userClientHandler);
                    }
                });
        bootstrap.connect(serverWrapper.getServerMeta().getIp(), serverWrapper.getServerMeta().getPort()).sync();
        System.out.println("Client[" + serverWrapper.getServerMeta().getIp() + ":" + serverWrapper.getServerMeta().getPort() + "] start now.");
        return userClientHandler;
    }

    class ServerTimeCheckTask extends TimerTask {

        @Override
        public void run() {
            long now = System.currentTimeMillis();
            List<ServerWrapper<ServerMeta>> list;
            synchronized (serverLock) {
                list = RpcConsumer.this.serverList;
            }
            for (ServerWrapper<ServerMeta> serverWrapper : list) {
                Long lastResponseTime = serverWrapper.getLastResponseTime();
                if (now - lastResponseTime >= 5 * 1000) {
                    serverWrapper.setLastRequestTime(0L);
                    serverWrapper.setLastResponseTime(0L);
                    serverWrapper.setLastHandlerTime(0L);
                }
            }
        }
    }
}
